
#include <algorithm>
#include <stdio.h>
#include <string>

#include "chordgod.h"
#include "undertopomgr.h"
#include "peerloger.h"
#include "config.h"
#include "chordnode.h"
#include "random.h"
#include "chordprotocol.h"
#include "chordkeyinfo.h"

#define DEBUG_PEER 1
#define DEBUG_FINGER 1
#define DEBUG_KEY 1
#define FINGER_TABLE_SIZE 32

class __CHORDNODE_ID_COMP{
	public:
		bool operator() ( ChordNode *n1, ChordNode *n2 ){
			return ( (n1->getChordNodeID()) < (n2->getChordNodeID()) );
		}
};

class __CHORDNODE_IDKEY_COMP{
	public:
		bool operator() ( chord_key_t key, ChordNode *n ){
			return ( key < (n->getChordNodeID()) );
		}
};

class __CHORDNODE_LOAD_COMP{
	public:
		bool operator() ( ChordNode *n1, ChordNode *n2 ){
			return ( (n1->getProtocolByID(0))->getNumberOfKey() < (n2->getProtocolByID(0))->getNumberOfKey() );
		}
};

ChordGod* ChordGod::instance_ = NULL;

bool ChordGod::create()
{
	NEW_OBJECT( ChordGod, instance_, () );
	return true;
}

bool ChordGod::init()
{
	assert_or_retv( config.chordgod_.number>0, false );

	ChordNetInfo info;
	chord_nets_.reserve( config.chordgod_.number + 1 );
	for( int i=0; i<config.chordgod_.number; i++ ){
		info.net_id_ = i;
		chord_nets_.push_back( info );
	}

	if ( config.chordgod_.reserve ){
		int node_no = UnderTopoMgr::instance().networkSize();
		assert_or_retv( node_no>0, false );
		TChordNetInfoVector::iterator itr = chord_nets_.begin(),
			itre = chord_nets_.end();
		while(itr!=itre){
			ChordNetInfo& info = *itr;
			info.nodes_.reserve( (size_t)node_no );
			itr++;
		}
	}

	//create key tuples for each chordnet
	{
		int number_of_key = config.chordgod_.number_of_key;
		TChordNetInfoVector::iterator itr = chord_nets_.begin(),
			itre = chord_nets_.end();
		while(itr!=itre){
			ChordNetInfo& info = *itr;
			info.key_tuples_ = new KeyTuple[number_of_key];
			info.number_of_keys_ = number_of_key;
			fill( info );
			itr++;
		}
	}

	PeerLoger::instance().logDebug( 0, "ChordGod init ok, number of chordnet:%d", chord_nets_.size() );

	return true;

}

bool ChordGod::initChord()
{
	if (config.chordgod_.initial_chord_size <= 0) return false;

	assert_or_retv( (uint32_t)config.chordgod_.initial_chord_size 
			<= UnderTopoMgr::instance().networkSize(), false );

	chord_nodes_.resize(config.chordgod_.number);
	{
		std::vector<TChordNodeVector>::iterator itr = chord_nodes_.begin(),
			itre = chord_nodes_.end();
		for (; itr!=itre; itr++) {
			itr->reserve(config.chordgod_.initial_chord_size/config.chordgod_.number+1);
		}
	}

	TChordNodeVector chord_nodes_temp;
	chord_nodes_temp.resize(config.chordgod_.initial_chord_size);
	for (int i=0; i<config.chordgod_.initial_chord_size; i++) {
		chord_nodes_temp[i] = (static_cast<ChordNode*>(UnderTopoMgr::instance().node(i)));
	}

	for (int j=0; j<config.chordgod_.number;j++) {
		for (int i=config.chordgod_.initial_chord_size*j/config.chordgod_.number; 
				i<config.chordgod_.initial_chord_size*(j+1)/config.chordgod_.number; 
				i++) {
			chord_nodes_temp[i]->setStartupNetID(j);
		}
	}

	std::sort( chord_nodes_temp.begin(), chord_nodes_temp.end(), __CHORDNODE_ID_COMP() );

	TChordNodeVector::iterator itr, itr_next, itr_begin, itr_end;
	for (itr = chord_nodes_temp.begin(); itr != chord_nodes_temp.end(); itr++) {
		chord_nodes_[(*itr)->startup_net_id_].push_back(*itr);
	}

	for (int i=0; i<config.chordgod_.number; i++) {
		itr_begin = chord_nodes_[i].begin();
		itr_end = chord_nodes_[i].end();
		if (itr_begin == itr_end) continue;

		itr = itr_next = itr_begin;
		NEW_VAR_OBJECT( ChordProtocol, p, ((*itr), i) );
		(*itr)->protocols_.push_back( p );
		registerToChordNet(i, (*itr)->node_id_);

		for (; itr != itr_end; itr++) {
			itr_next++;
			if (itr_next == itr_end) {
				(*(*itr)->protocols_.begin())->setSuccessor((*itr_begin)->node_id_, (*itr_begin)->chord_nid_);
				(*(*itr_begin)->protocols_.begin())->setPredecessor((*itr)->node_id_, (*itr)->chord_nid_);
			}
			else {
				(*(*itr)->protocols_.begin())->setSuccessor((*itr_next)->node_id_, (*itr_next)->chord_nid_);
				NEW_VAR_OBJECT( ChordProtocol, p, ((*itr), i) );
				(*itr_next)->protocols_.push_back( p );
				registerToChordNet(i, (*itr_next)->node_id_);
				(*(*itr_next)->protocols_.begin())->setPredecessor((*itr)->node_id_, (*itr)->chord_nid_);
			}
		}
	}

	assignKeys();
	dumpLoad();
/*
#ifdef DEBUG_PEER
	dumpChordNet();
#endif

	// >>> initialize finger table ...
	int i, j, index, last_index;
#ifdef DEBUG_FINGER
	FILE *outfile = fopen("logs//initial_finger_tables.out", "w");
#endif
	for (i=0; i<config.chordgod_.number; i++) {
#ifdef DEBUG_FINGER
		fprintf( outfile, "finger tables:\nnet_id:%u\n", i );
#endif
		itr_begin = chord_nodes_[i].begin();
		itr_end = chord_nodes_[i].end();

		itr = itr_begin;
		itr++;
		if (itr == itr_end) { // only one node in the chord net
#ifdef DEBUG_FINGER
			fprintf( outfile, "current_nid:%u@%u\n", (*itr)->chord_nid_, (*itr)->node_id_ );
#endif

			for (j=0; j<FINGER_TABLE_SIZE; j++) {
				(*(*itr)->protocols_.begin())->fingers_[j].node_id_ = (*itr)->node_id_;
				(*(*itr)->protocols_.begin())->fingers_[j].chord_nid_ = (*itr)->chord_nid_;
				(*(*itr)->protocols_.begin())->fingers_[j].last_update_ = 0;
#ifdef DEBUG_FINGER
				fprintf( outfile, "index:%d suc_nid:%u@%u\n", j, (*itr)->chord_nid_, (*itr)->node_id_ );
#endif
			}
			continue;
		}

		for (itr = itr_begin; itr != itr_end; itr++) {
#ifdef DEBUG_FINGER
			fprintf( outfile, "current_nid:%u@%u\n", (*itr)->chord_nid_, (*itr)->node_id_ );
#endif
			itr_next = itr;
			itr_next++;
			if (itr_next == itr_end) itr_next = itr_begin;
			last_index = -1;
			while (itr_next != itr) {
				index = ChordProtocol::keyToFingerIndex((*itr)->chord_nid_, (*itr_next)->chord_nid_);
				for (j = last_index+1; j <= index; j++) {
					(*(*itr)->protocols_.begin())->fingers_[j].node_id_ = (*itr_next)->node_id_;
					(*(*itr)->protocols_.begin())->fingers_[j].chord_nid_ = (*itr_next)->chord_nid_;
					(*(*itr)->protocols_.begin())->fingers_[j].last_update_ = 0;	
#ifdef DEBUG_FINGER
					fprintf( outfile, "index:%d suc_nid:%u@%u\n", j, (*itr_next)->chord_nid_, (*itr_next)->node_id_ );
#endif
				}
				itr_next++;
				if (itr_next == itr_end) itr_next = itr_begin;
				last_index = index;
			}
		}
	}
#ifdef DEBUG_FINGER
	fclose(outfile);
	outfile = NULL;
#endif
*/
	return true;

}

ChordNode* ChordGod::getNodeInChordNet( chord_net_id_t net_id, node_id_t excluded )
{
	assert_or_retv( (net_id>=0 && net_id<numberOfChordNet()) , false );

	TNodeIDVector& nodes = chord_nets_[net_id].nodes_;

	if ( nodes.empty() ) return NULL;
	if ( (nodes.size()==1) && (nodes[0]==excluded) ) return NULL;

	node_id_t selected;
	do{
		selected = nodes[ (size_t)rng.uniform(0, (int)nodes.size()) ];
	} while( selected==excluded );

	Node* n = UnderTopoMgr::instance().node( selected );
	assert_or_retv( (NULL!=n), NULL );

	return static_cast<ChordNode*>( n );

}

bool ChordGod::registerToChordNet( chord_net_id_t net_id, node_id_t node_id )
{
	assert_or_retv( (net_id>=0 && net_id<numberOfChordNet()) , false );

	//TODO: duplicate check
	chord_nets_[net_id].nodes_.push_back( node_id );
	return true;
}

bool ChordGod::unregisterToChordNet( chord_net_id_t net_id, node_id_t node_id )
{
	assert_or_retv( (net_id>=0 && net_id<numberOfChordNet()) , false );
	ChordNetInfo& net_info = chord_nets_[net_id];
	TNodeIDVector::iterator itr = std::find( net_info.nodes_.begin(), net_info.nodes_.end(), node_id );
	if ( itr==net_info.nodes_.end() ) return false;
	net_info.nodes_.erase( itr );
	//(*itr) = *(net_info.nodes_.end()-1);
	//net_info.nodes_.pop_back();
	return true;
}

bool ChordGod::unregisterKeyToChordNet( chord_net_id_t net_id, chord_key_t key )
{
	assert_or_retv( net_id<chord_nets_.size(), false);
	ChordNetInfo& info = chord_nets_[net_id];
	assert_or_retv( NULL!=info.key_tuples_, false);

	KeyTuple* kt = std::find( info.key_tuples_, info.key_tuples_+info.number_of_keys_, key );
	if ( kt!=info.key_tuples_+info.number_of_keys_ ){
		kt->setOnline(false);
		return true;
	}
	return false;
}

bool ChordGod::isKeyOnline( chord_net_id_t net_id, chord_key_t key )
{
	assert_or_retv( net_id<chord_nets_.size(), false);
	ChordNetInfo& info = chord_nets_[net_id];
	assert_or_retv( NULL!=info.key_tuples_, false);

	KeyTuple* kt = std::find( info.key_tuples_, info.key_tuples_+info.number_of_keys_, key );
	if ( kt!=info.key_tuples_+info.number_of_keys_ ){
		return kt->isOnline();
	}
	return false;
}

ChordNetInfo& ChordGod::getNetInfo( chord_net_id_t net_id )
{
	return chord_nets_[net_id];
}

void ChordGod::destroy()
{
	RELEASE_OBJ( instance_ )
}

uint32_t ChordGod::getKeySpaceSize( chord_net_id_t net_id )
{
	return (uint32_t)std::numeric_limits<chord_key_t>::max();
}

void ChordGod::dumpChordNet()
{
	FILE *outfile = fopen("logs//chordnet.out", "w");
	if ( NULL==outfile ){
		fprintf( stderr, "ChordGod::dumpChordNet cannot open file\n");
		return;
	}
	TChordNetInfoVector::iterator itr = chord_nets_.begin(),
		itre = chord_nets_.end();
	for ( ; itr!=itre; itr++ ){
		ChordNetInfo& net_info = *itr;
		fprintf( outfile, "net_id:%u online:%u\n", net_info.net_id_, (uint32_t)net_info.nodes_.size() );
		TNodeIDVector::iterator itr_id = net_info.nodes_.begin(),
			itre_id = net_info.nodes_.end();
		for ( ; itr_id!=itre_id; itr_id++ ){
			ChordNode *cn = static_cast<ChordNode*>(UnderTopoMgr::instance().node( *itr_id ) );
			ChordProtocol* p = cn->getProtocolByID( net_info.net_id_ );
			ChordNodeInfo& suc = p->getSuccessor();
			ChordNodeInfo& pre = p->getPredecessor();
			std::string str_fingers, str_keys;
			p->fingerString( str_fingers );
			p->keyString( str_keys );
			fprintf( outfile, "node:%u@%u succ:%u@%u pre:%u@%u %s %s\n", cn->getChordNodeID(), cn->getNodeID(), 
					suc.chord_nid_, suc.node_id_, pre.chord_nid_, pre.node_id_, str_fingers.c_str(), str_keys.c_str() );

		}
		fprintf(outfile, "\n");
	}
	fclose(outfile);
	outfile = NULL;
}

void ChordGod::dumpLoad()
{
	if (config.chordgod_.number > 1) {
		fprintf( stderr, "In the simulation of load balance, the number of chord nets shoud be 1!\n" );
		return;
	}

	FILE *outfile = fopen("logs//load.out", "w");
	if ( NULL==outfile ){
		fprintf( stderr, "ChordGod::dumpLoad cannot open file\n");
		return;
	}

	TChordNetInfoVector::iterator itr = chord_nets_.begin(),
		itre = chord_nets_.end();
	for ( ; itr!=itre; itr++ ){
		ChordNetInfo& net_info = *itr;
		TNodeIDVector::iterator itr_id = net_info.nodes_.begin(),
			itre_id = net_info.nodes_.end();
		for ( ; itr_id!=itre_id; itr_id++ ){
			ChordNode *cn = static_cast<ChordNode*>(UnderTopoMgr::instance().node( *itr_id ) );
			ChordProtocol* p = cn->getProtocolByID( net_info.net_id_ );
			fprintf( outfile, "node:%u@%u keytotal:%u\n", cn->getChordNodeID(), cn->getNodeID(), 
					p->getNumberOfKey() );
		}
	}
	fclose(outfile);
	outfile = NULL;
}

void ChordGod::generateKeys(int power_of_n, int number_of_key )
{
	uint32_t rand;

	random_key_list_.resize(power_of_n);

	for (int i = 0; i < power_of_n; i++) {
		random_key_list_[i].resize(number_of_key);
		for (int j = 0; j < number_of_key;) {
			rand = rng.uniform(0, std::numeric_limits<int>::max());
			if (std::find(random_key_list_[i].begin(), random_key_list_[i].end(), rand)
					== random_key_list_[i].end()) {
				random_key_list_[i][j] = rand;
				j++;
			}
		}
	}
}

void ChordGod::fill_seq( ChordNetInfo& info )
{
        assert( NULL!=info.key_tuples_ );
        size_t power_of_n = 1;
        for ( int i=0; i<info.number_of_keys_; i++ ){
                info.key_tuples_[i].resize( power_of_n );
                info.key_tuples_[i][0] = i;
        }
}

void ChordGod::fill( ChordNetInfo& info )
{
	assert( NULL!=info.key_tuples_ );
	generateKeys(config.chordgod_.power_of_n, config.chordgod_.number_of_key);

	for ( int i = 0; i < config.chordgod_.number_of_key; i++ ){
		info.key_tuples_[i].resize( config.chordgod_.power_of_n );
		for ( int j = 0; j < config.chordgod_.power_of_n; j++){
			info.key_tuples_[i][j] = random_key_list_[j][i];
		}
	}
}

void ChordGod::assignKeys()
{
	if (config.chordgod_.number > 1) {
		fprintf( stderr, "In the simulation of load balance, the number of chord nets shoud be 1!\n" );
		return;
	}
#ifndef CHORD_POWER_OF_N
	assert(config.chordgod_.power_of_n==1);
		fprintf( stderr, "Compiled with power_of_n disabled, but n is larger than 1\n" );
		return;
#endif 

	TChordNodeVector chord_nodes_temp;
	chord_nodes_temp.resize(config.chordgod_.power_of_n);

	TChordNodeVector::iterator itr;
	uint32_t i;
	int j, index;

	for (i = 0; i < config.chordgod_.number_of_key; i++) {
		//chord_nodes_temp.clear();
		for (j = 0; j < config.chordgod_.power_of_n; j++) {
			itr = std::upper_bound( chord_nodes_[0].begin(), chord_nodes_[0].end(), chord_nets_[0].key_tuples_[i][j], __CHORDNODE_IDKEY_COMP() );	
			if (itr == chord_nodes_[0].end()) {
				itr = chord_nodes_[0].begin();
			}
			chord_nodes_temp[j] = (*itr);
		}
		
		itr = std::min_element( chord_nodes_temp.begin(), chord_nodes_temp.end(), __CHORDNODE_LOAD_COMP() );
		//std::sort( chord_nodes_temp.begin(), chord_nodes_temp.end(), __CHORDNODE_ID_COMP() );
		//itr = chord_nodes_temp.begin();
		index = std::distance(chord_nodes_temp.begin(), itr);
		NEW_VAR_OBJECT( ChordKeyInfo, ck, ( &(chord_nets_[0].key_tuples_[i]), index ) );
		(*((*itr)->protocols_.begin()))->addKey(ck);
	}
}

void ChordGod::startMerge()
{
	chord_net_id_t net_id = chord_nets_.size();
	ChordNetInfo info;
	info.net_id_ = net_id;
	chord_nets_.push_back( info );

	TChordNetInfoVector::iterator itr = chord_nets_.begin(),
		itre = chord_nets_.end()-1;//the last element is the new net
	for ( ; itr!=itre; itr++ ){
		ChordNetInfo& n_info = *itr;
		ChordNode *n = getNodeInChordNet( n_info.net_id_, invalid_node_id );
		if ( n!=NULL ){
			assert( n->isOnline() );
			fprintf( stderr, "\nmerge %u to %u choose node :%u\n", n_info.net_id_, net_id, n->getNodeID() );
			n->merge( n_info.net_id_, net_id );

		}
	}
}

